package com.tianji.login.service.Impl;

;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tianji.common.Result;
import com.tianji.login.mapper.SysUserMapper;
import com.tianji.login.pojo.SysUser;
import com.tianji.login.service.ESService;
import com.tianji.login.service.SysUserService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * <p>
 * 系统用户 服务实现类
 * </p>
 *
 * @author 陈佳
 * @since 2024-09-30
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class SysUserServiceImpl extends ServiceImpl<SysUserMapper, SysUser> implements SysUserService {

    private final SysUserMapper sysUserMapper;

    private static final int PAGE_SIZE = 50000;

    // 创建适合IO密集型任务的线程池
    private final ExecutorService createOrderExecutor = new ThreadPoolExecutor(
            // 核心线程数设置为CPU核心数的2倍
            Runtime.getRuntime().availableProcessors() * 2,
            // 最大线程数设置为CPU核心数的3倍
            Runtime.getRuntime().availableProcessors() * 3,
            // 线程空闲时间
            60L, TimeUnit.SECONDS,
            // 阻塞队列容量
            new LinkedBlockingQueue<>(100),
            // 拒绝策略，直接由调用线程执行任务
            new ThreadPoolExecutor.CallerRunsPolicy()
    );


    private final ESService esService;

    private static final int MAX_TRY_COUNT = 3;





    @Override
    public Result createOrder() {
        //查询总数，计算分页
        Long total = sysUserMapper.selectCount(null);
        int totalPage = (int)Math.ceil((double) total/PAGE_SIZE);
        List<Integer> failPages = new ArrayList<>();

        //初始化闭锁
        CountDownLatch latch = new CountDownLatch(totalPage);
        for (int i = 0; i < totalPage; i++) {
            final int index = i;
            createOrderExecutor.submit((()->{
                try {
                    //数据导入es,并且带入失败重试不回滚逻辑
                    bulkInsertWuthRetry(index);
                } catch (Exception e) {
                    //记录失败页码
                    synchronized (failPages) {
                        failPages.add(index);
                    }
                    e.printStackTrace();
                }finally {
                    //每个任务结束之后计数器减一
                    latch.countDown();
                }
            }));
        }
        try {
            // 等待所有线程执行完成
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();  // 重新设置线程的中断状态
            return Result.failure("线程被中断，导入失败");
        }
        createOrderExecutor.shutdown();

        if(!failPages.isEmpty()) {
            return Result.failure("部分数据导入失败，失败页码："+failPages);
        }
        return Result.success("导入千万级客户信息到es成功");
    }

    private void bulkInsertWuthRetry(int index) throws IOException {
        for (int i = 0; i < MAX_TRY_COUNT; i++) {
            try {
                Page<SysUser> sysUserPage = sysUserMapper.selectPage(new Page<>(index, PAGE_SIZE), null);
                esService.bulkInsert(sysUserPage.getRecords());
            }catch (Exception e) {
                if(i == MAX_TRY_COUNT-1){
                    log.error("第 {} 页数据导入失败，达到了最大重试次数 {}",index,MAX_TRY_COUNT);
                    throw e;
                }
                log.error("第 {} 页数据导入失败，重试中....(第{}次重试)",index,i);
            }
        }
    }
}
